package com.flink.window;

import com.flink.datasource.UserSource;
import com.flink.entity.User;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.Duration;
import java.util.HashSet;
import java.util.Set;

import static sun.misc.Version.print;

/**
 * 描述:
 * TODO 增量聚合函数和全窗口函数结合使用
 *
 * @author yanzhengwu
 * @create 2022-07-31 20:48
 */
public class WindowProcessAndAggregateFunction {

    public static void main(String[] args) throws Exception {

        //声明执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //测试为了保证元素顺序并行度设置为1，可以理解为单线程执行
        env.setParallelism(1);
        //设置水位线生成的间隔 这里给的是100毫秒 ,flink 默认是200毫秒 ，flink 可以达到毫秒级别的效率
        env.getConfig().setAutoWatermarkInterval(100);


        //TODO 无序流的watermark生成策略
        DataStream<User> stream = env.addSource(new UserSource())       //生成水位线和时间戳的策略对象
                .assignTimestampsAndWatermarks(
                        //返回一个具体的策略对象(TODO 这里是乱序流的处理方法，给了一个延迟2秒的策略，也可由理解为 数据延迟多长时间能够全部到位)
                        WatermarkStrategy.<User>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                                //返回策略对象的具体实现
                                .withTimestampAssigner(new SerializableTimestampAssigner<User>() {
                                    /**
                                     * 此方法为指定以事件时间的具体时间戳字段
                                     *
                                     * @param element
                                     * @param recordTimestamp
                                     * @return 返回的则是一个毫秒数的时间戳
                                     */
                                    @Override
                                    public long extractTimestamp(User element, long recordTimestamp) {
                                        return element.getTimestamp();
                                    }
                                }));
        //打印数据和 计算结果进行区分
        stream.print("data==>");
        //数据倾斜就是为了在一定场景下必须等所有数据都到期了在进行计算
        stream.keyBy(data -> true)
                .window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .aggregate(new MyAggregateFunction(), new MyWindowProcessFunction())
                .print();

        env.execute();
    }

    //自定义增量聚合函数
    public static class MyAggregateFunction implements AggregateFunction<User, HashSet<String>, Long> {

        @Override
        public HashSet<String> createAccumulator() {
            return new HashSet<>();
        }

        @Override
        public HashSet<String> add(User value, HashSet<String> accumulator) {
            accumulator.add(value.getName());
            return accumulator;
        }

        @Override
        public Long getResult(HashSet<String> accumulator) {
            return (long) accumulator.size();
        }

        @Override
        public HashSet<String> merge(HashSet<String> a, HashSet<String> b) {
            return null;
        }
    }

    //自定义一个全窗口函数
    public static class MyWindowProcessFunction extends ProcessWindowFunction<Long, String, Boolean, TimeWindow> {

        @Override
        public void process(Boolean aBoolean, Context context, Iterable<Long> elements, Collector<String> out) throws Exception {
            //前面做了聚合窗口函数只返回了一个访问数量 所以这里去取一个即可里面也只有一个值
            long size = elements.iterator().next();
            long start = context.window().getStart();
            long end = context.window().getEnd();

            out.collect("窗口起始点："
                    + new Timestamp(start)
                    + ";" + "窗口结束点："
                    + new Timestamp(end)
                    + "访问数量：" + size);
        }
    }
}
